package com.qgs.secsgem.mq.producer.component;

import com.alibaba.fastjson.JSONObject;
import com.qgs.secsgem.common.utils.CommonFunction;
import com.qgs.secsgem.mq.entity.MQMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Random;
import java.util.UUID;

@Component
public class RabbitSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;


    private static final Logger logger = LoggerFactory.getLogger(RabbitSender.class);


    final RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback(){
        //rabbitTemplate.setReturnCallback 会调用这个接口
        //第二种publisher-returns模式，该模式会在消息没有被路由到queues时将消息返回
        @Override
        public void returnedMessage(ReturnedMessage returned) {
            org.springframework.amqp.core.Message message = returned.getMessage();
            logger.error("消息未能成功路由到指定queues : return--message:" + new String(message.getBody()) + ",replyCode:" + returned.getReplyCode()
                    + ",replyText:" + returned.getReplyText() + ",exchange:" + returned.getExchange() + ",routingKey:" + returned.getRoutingKey());
        }
    };


    /**
     * rabbitTemplate.setConfirmCallback() 会调用这个接口。
     * 确认消息 的回调监听接口，，用于确认消息是否被broker所收到(也就是 是否投递到exchange（不能保证是否投递到queue）)。
     * 当消息被broker 成功接收到 后 会调用这个方法
     * */
    /**
     * 参数：
     *      CorrelationData  作为一个唯一标识（ 生产者 消费者 消息确认的唯一标识 ）
     *      b                broker是否落盘成功
     *      s                失败的一些异常信息会送
     * */
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         * 参数：
         *      CorrelationData  作为一个唯一标识（ 生产者 消费者 消息确认的唯一标识 ）
         *      ack              broker是否落盘成功
         *      s                失败的一些异常信息会送
         * */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            //如果RabbitMQ 因为自身内部错
            //误导致消息丢失，就会发送一条nack (Basic . Nack ）命令，生产者应用程序同样可以在回调
            //方法中处理该nack 命令。
            if (ack) {
                logger.info("消息 transactionId: [" + correlationData.getId() + "] 成功发送到指定ExChange");
            } else {
                logger.info("消息 transactionId: [" + correlationData.getId() + "] 发送到ExChange失败,原因 : " + cause);
            }
        }
    };


    /**
     *  @Description 向rabbitmq中发送消息（topic模式）
     *  @Param
     *          message     具体的消息内容
     *          properties  额外的附加属性
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17 
     **/
    public void sendTopic(Object message, Map<String,Object> properties, String routingKey) throws Exception{
        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);

        //唯一标识
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());

        //设置确认 confirm
        rabbitTemplate.setConfirmCallback(confirmCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 成功 之后调用下面的方法。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                System.err.println("posted todo:"+message);
                //message.getMessageProperties()  下 可以进行各种 SET 属性
                return message;
            }
        };
        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend("rc_exchange",
                routingKey,
                msg,
                mpp,
                cd
        );
    }


    /**
     *  @Description 向rabbitmq中发送消息（direct模式）
     *  @Param
     *          mqMessage   通用的消息格式
     *          properties  额外的附加属性,放在Message Header里
     *          exchange    exchangeName
     *          routingKey  路由键
     *          expiration  设置这条消息的过期时间，如 6000 =》 6s ，6s后未被消费就过期
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17
     **/
    public void sendDirect(MQMessage mqMessage, Map<String,Object> properties, String exchange, String routingKey, String expiration) throws Exception{
        byte[] message = JSONObject.toJSONBytes(mqMessage);

        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag

        //content_type 默认 就是application/octet-stream 不是json格式的
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);


        //唯一标识
        String transitionId = mqMessage.getHeader().getEquipmentId()+"_"+ CommonFunction.getNowDate(2)+"_"+CommonFunction.getUUID(5);
        CorrelationData cd = new CorrelationData(transitionId);


        //设置确认 confirm(),投递到Exchange 成功的时候 会调用下面这个回调方法
        //rabbitTemplate.setConfirmCallback(confirmCallback);   //ok


        //注意 有关rabbitTemplate 配置 的信息 不能写在这里，
        // 因为 每次外部REST接口过来（不同线程） 调用sendDirect方法的时候  都会setConfirmCallback(new ConfirmBackSender());
        //而rabbitmq只允许 设置 一次 ConfirmBack；这里new 了这么多 肯定不对
        //rabbitTemplate.setConfirmCallback(new ConfirmBackSender());  //这样不行
        rabbitTemplate.setConfirmCallback(confirmCallback);          //这样可以,confirmBackSender就是把回调处理类 独立出去了 ，和写在本类中的returnCallback 做一个区别

        //设置确认 returnedMessage();  当未投递到queue的时候 会调用下面这个方法。
        //注意 ：如果要设置setReturnCallback 那么 必须配置spring.rabbitmq.publisher-returns=true,spring.rabbitmq.template.mandatory=true
        rabbitTemplate.setReturnsCallback(returnsCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 之前调用下面的方法，也就是对消息进行加工。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
//                System.err.println("posted todo:"+message);
//                //message.getMessageProperties()  下 可以进行各种 SET 属性
//                Random random = new Random();
//                int i = random.nextInt(10);
//                message.getMessageProperties().setPriority(i);
                //过期时间： 单位毫秒 (注意：这些消息 只有在队列队首的 时候过期了才会移除，否则是不会被移除的)
                message.getMessageProperties().setExpiration(expiration);
                return message;
            }
        };


        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend(exchange,
                routingKey,
                msg,
                mpp,
                cd
        );
    }



    public void sendDirect(MQMessage mqMessage, String exchange, String routingKey) throws Exception{

        byte[] message = JSONObject.toJSONBytes(mqMessage);

        String transitionId = mqMessage.getHeader().getTransactionId();
        CorrelationData cd = new CorrelationData(transitionId);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnsCallback(returnsCallback);
        //rabbitTemplate.send 此方法发送的消息 是不含java相关信息的，rabbitTemplate.convertAndSend是包含java相关信息的
        org.springframework.amqp.core.Message mqsMessage = new org.springframework.amqp.core.Message(message);
        rabbitTemplate.send(exchange,routingKey,mqsMessage,cd);

//        MessageHeaders messageHeaders = new MessageHeaders(properties);
//        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);
//        rabbitTemplate.convertAndSend(exchange,
//                routingKey,
//                msg,
//                cd
//        );

    }


    /**
     *  @Description 回复EAP的request（向rabbitmq中发送消息（direct模式））
     *  @Param
     *          mqMessage   通用的消息格式
     *          properties  额外的附加属性,放在Message Header里
     *          exchange    exchangeName
     *          routingKey  路由键
     *          expiration  设置这条消息的过期时间，如 6000 =》 6s ，6s后未被消费就过期
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17
     **/
    public void reply(MQMessage mqMessage, Map<String,Object> properties, String exchange, String routingKey, String expiration) throws Exception{
        byte[] message = JSONObject.toJSONBytes(mqMessage);

        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag

        //content_type 默认 就是application/octet-stream 不是json格式的
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);


        //唯一标识
        String transitionId = mqMessage.getHeader().getEquipmentId()+"_"+ CommonFunction.getNowDate(2)+"_"+CommonFunction.getUUID(5);
        CorrelationData cd = new CorrelationData(transitionId);


        //设置确认 confirm(),投递到Exchange 成功的时候 会调用下面这个回调方法
        //rabbitTemplate.setConfirmCallback(confirmCallback);   //ok


        //注意 有关rabbitTemplate 配置 的信息 不能写在这里，
        // 因为 每次外部REST接口过来（不同线程） 调用sendDirect方法的时候  都会setConfirmCallback(new ConfirmBackSender());
        //而rabbitmq只允许 设置 一次 ConfirmBack；这里new 了这么多 肯定不对
        //rabbitTemplate.setConfirmCallback(new ConfirmBackSender());  //这样不行
        rabbitTemplate.setConfirmCallback(confirmCallback);          //这样可以,confirmBackSender就是把回调处理类 独立出去了 ，和写在本类中的returnCallback 做一个区别

        //设置确认 returnedMessage();  当未投递到queue的时候 会调用下面这个方法。
        //注意 ：如果要设置setReturnCallback 那么 必须配置spring.rabbitmq.publisher-returns=true,spring.rabbitmq.template.mandatory=true
        rabbitTemplate.setReturnsCallback(returnsCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 之前调用下面的方法，也就是对消息进行加工。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
//                System.err.println("posted todo:"+message);
//                //message.getMessageProperties()  下 可以进行各种 SET 属性
//                Random random = new Random();
//                int i = random.nextInt(10);
//                message.getMessageProperties().setPriority(i);
                //过期时间： 单位毫秒 (注意：这些消息 只有在队列队首的 时候过期了才会移除，否则是不会被移除的)
                message.getMessageProperties().setExpiration(expiration);
                return message;
            }
        };


        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend(exchange,
                routingKey,
                msg,
                mpp,
                cd
        );
    }
    public void reply(MQMessage mqMessage, Map<String,Object> properties, String exchange, String routingKey) throws Exception{
        byte[] message = JSONObject.toJSONBytes(mqMessage);

        MessageHeaders messageHeaders = new MessageHeaders(properties);

        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);

        String transitionId = mqMessage.getHeader().getEquipmentId()+"_"+ CommonFunction.getNowDate(2)+"_"+CommonFunction.getUUID(5);
        CorrelationData cd = new CorrelationData(transitionId);

        rabbitTemplate.setConfirmCallback(confirmCallback);


        rabbitTemplate.setReturnsCallback(returnsCallback);

        MessagePostProcessor mpp = new MessagePostProcessor(){

            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                return message;
            }
        };


        rabbitTemplate.convertAndSend(exchange,
                routingKey,
                msg,
                mpp,
                cd
        );
    }
    //todo sendDirect应该改成 sendCommon（加上T3超时校验），还要加一个reply() 方法 专门用于 eapRequest 回复用





    /**
     *  @Description 向rabbitmq中发送消息（direct模式）
     *  @Param
     *          message     具体的消息内容
     *          properties  额外的附加属性
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17
     **/
    public void sendDirect(Object message, Map<String,Object> properties,String routingKey,String exchange) throws Exception{
        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag

        //content_type 默认 就是application/octet-stream 不是json格式的
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);


        //唯一标识
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());


        //设置确认 confirm(),投递到Exchange 成功的时候 会调用下面这个回调方法
        //rabbitTemplate.setConfirmCallback(confirmCallback);   //ok


        //注意 有关rabbitTemplate 配置 的信息 不能写在这里，
        // 因为 每次外部REST接口过来（不同线程） 调用sendDirect方法的时候  都会setConfirmCallback(new ConfirmBackSender());
        //而rabbitmq只允许 设置 一次 ConfirmBack；这里new 了这么多 肯定不对
        //rabbitTemplate.setConfirmCallback(new ConfirmBackSender());  //这样不行
        rabbitTemplate.setConfirmCallback(confirmCallback);          //这样可以,confirmBackSender就是把回调处理类 独立出去了 ，和写在本类中的returnCallback 做一个区别

        //设置确认 returnedMessage();  当未投递到queue的时候 会调用下面这个方法。
        //注意 ：如果要设置setReturnCallback 那么 必须配置spring.rabbitmq.publisher-returns=true,spring.rabbitmq.template.mandatory=true
        rabbitTemplate.setReturnsCallback(returnsCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 之前调用下面的方法，也就是对消息进行加工。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                System.err.println("posted todo:"+message);
                //message.getMessageProperties()  下 可以进行各种 SET 属性
                Random random = new Random();
                int i = random.nextInt(10);
                message.getMessageProperties().setPriority(i);
                //过期时间： 单位毫秒 (注意：这些消息 只有在队列队首的 时候过期了才会移除，否则是不会被移除的)
                message.getMessageProperties().setExpiration("6000");
                return message;
            }
        };


        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend(exchange,
                routingKey,
                msg,
                mpp,
                cd
        );
    }
}
